package com.inf.dao;

import com.inf.utils.HBaseUtil;
import com.inf.utils.PhoenixUtil;
import com.inf.utils.Tools;
import com.inf.vo.BaseVo;
import com.inf.vo.ParamsPartnerVo;
import com.inf.vo.ResultPartnerVo;
import org.apache.commons.io.IOUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.filter.CompareFilter;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.RegexStringComparator;
import org.apache.hadoop.hbase.filter.RowFilter;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Repository;

import java.io.IOException;
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.Statement;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.*;

/**
 * Created by zhangbn on 2018/11/14.
 */
@Repository
public class PersonRepoDao {


    Logger log = LoggerFactory.getLogger(PersonRepoDao.class);

    /**
     * 从原始数据中查询同行人
     *
     * @param tableName
     * @param paramVo
     * @return
     * @throws IOException
     */
    public List<ResultPartnerVo> queryPartnerByRangeAndRowkeyInOriginalView(String tableName, ParamsPartnerVo paramVo, String rowkey) throws IOException {
        List<ResultPartnerVo> partnerRst = new ArrayList<>();
        Set<String> cameraSet = new HashSet<>();
        if (StringUtils.isNotEmpty(paramVo.getCameraIds())) {
            String[] cameraArr = paramVo.getCameraIds().split(",");
            List<String> cameraList = Arrays.asList(cameraArr);
            cameraSet = new HashSet<>(cameraList);
        }
        Statement statement = null;
        ResultSet result = null;
        try {
            List<BaseVo> uuidMatchedVoList = new ArrayList<BaseVo>();
            String cameraIdSql = Tools.cameraIdsMakeStrForSql(paramVo.getCameraIds());
            DateTimeFormatter format = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
            DateTimeFormatter format_use = DateTimeFormatter.ofPattern("yyMMddHHmmss");
            LocalDateTime startTime = LocalDateTime.parse(paramVo.getBeginTime(), format);
            LocalDateTime endTime = LocalDateTime.parse(paramVo.getEndTime(), format);
            String startStr = startTime.format(format_use);
            String endStr = endTime.format(format_use);
            String uuid = rowkey.substring(12);
            log.info("\n\n");
            log.info("show: 开始从原始数据中查找[" + paramVo.getBeginTime() + " - " + paramVo.getEndTime() + "]之间rowkey以uuid= " + uuid + "结尾的目标图片......");
            log.info("\n\n");
            Connection conn = PhoenixUtil.getConnection();
            statement = conn.createStatement();
            String sql = "select * from \"bigdata\".\"person\" where (\"rowkey\" between '180722' and '180806') and (\"rowkey\" like \"%1807220001214b4c81ec41c665824935\")";
//            if (StringUtils.isNotBlank(cameraIdSql)) {
//                sql = sql + " and (\"camera_id\" in " + cameraIdSql + ")";
//            }
            result = statement.executeQuery(sql);
            int sum = 0;
            while (result.next()) {
                String row = result.getString("rowkey");
                String pass_id = result.getString("pass_id");
                Double quality_score = result.getDouble("quality_score");
                String face_img_url = result.getString("face_img_url");
                String pass_time = result.getString("pass_time");
                String camera_id = result.getString("camera_id");
                BaseVo vo = new BaseVo(row, pass_id, pass_time, camera_id, quality_score, face_img_url);
                uuidMatchedVoList.add(vo);
                if (!cameraSet.isEmpty()) {
                    if (cameraSet.contains(camera_id)) {
                        uuidMatchedVoList.add(vo);
                    }
                } else {
                    uuidMatchedVoList.add(vo);
                }
                sum++;
            }
            result.close();
            statement.close();
            log.info("\n\n");
            log.info("show: 处理sum= " + sum + "条rowkey以uuid= " + uuid + "结尾的目标图片完成");
            log.info("show: 处理完[" + paramVo.getBeginTime() + " - " + paramVo.getEndTime() + "]之间的目标数据后：得到指定相机组的条件下，uuid=[" + uuid + "]的人共出现 uuidMatchedVoList.size= " + uuidMatchedVoList.size() + " 次");
            log.info("\n\n");
            if (uuidMatchedVoList.size() > 0) {
                Map<ResultPartnerVo, Integer> partnerMap = new HashMap<>();
                for (BaseVo base : uuidMatchedVoList) {
                    List<ResultPartnerVo> oriList = queryVoByRowKeyWithRange("bigdata:person", base.getUuid(), paramVo.getPreSec(), paramVo.getPostSec(), cameraSet);
                    //log.info("show: uuid=[" + base.getUuid() + "]的oriList.size()= " + oriList.size());
                    Set<ResultPartnerVo> oriSet = new HashSet<>(oriList);
                    log.info("show: uuid=[" + base.getUuid() + "]的oriSet.size()= " + oriSet.size());
                    for (ResultPartnerVo k : oriSet) {
                        if (partnerMap.containsKey(k)) {
                            partnerMap.put(k, partnerMap.get(k) + 1);
                        } else {
                            partnerMap.put(k, 1);
                        }
                    }
                }
                log.info("show: uuid= [" + uuid + "] 的可能同行人partnerMap.size()= " + partnerMap.size());
                Iterator<Map.Entry<ResultPartnerVo, Integer>> m_it = partnerMap.entrySet().iterator();
                while (m_it.hasNext()) {
                    Map.Entry<ResultPartnerVo, Integer> entry = m_it.next();
                    if (entry.getValue() >= 2) {
                        ResultPartnerVo kv = entry.getKey();
                        kv.setTimes(entry.getValue());
                        partnerRst.add(kv);
                    }
                }
                log.info("show: 与uuid= [" + uuid + "] 同行2次及以上的人partnerRst.size()= " + partnerRst.size());
            }
            return partnerRst;
        } catch (Exception e) {
            log.error("查询同行人出错!", e);
        } finally {
            try {
                statement.close();
                result.close();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        return partnerRst;
    }


    /**
     * 从原始数据中查询同行人
     *
     * @param tableName
     * @param paramVo
     * @return
     * @throws IOException
     */
    public List<ResultPartnerVo> queryPartnerByRangeAndRowkeyInOriginal(String tableName, ParamsPartnerVo paramVo, String rowkey) throws IOException {
        List<BaseVo> uuidMatchedVoList = new ArrayList<BaseVo>();
        Set<String> cameraSet = new HashSet<>();
        if (StringUtils.isNotEmpty(paramVo.getCameraIds())) {
            String[] cameraArr = paramVo.getCameraIds().split(",");
            List<String> cameraList = Arrays.asList(cameraArr);
            cameraSet = new HashSet<>(cameraList);
        }
        DateTimeFormatter format = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
        DateTimeFormatter format_use = DateTimeFormatter.ofPattern("yyMMddHHmmss");
        LocalDateTime startTime = LocalDateTime.parse(paramVo.getBeginTime(), format);
        LocalDateTime endTime = LocalDateTime.parse(paramVo.getEndTime(), format);
        String startStr = startTime.format(format_use);
        String endStr = endTime.format(format_use);
        String uuid = rowkey.substring(12);
        Table table = HBaseUtil.connection.getTable(TableName.valueOf(tableName));
        Scan scan = new Scan();
        scan.setStartRow(Bytes.toBytes(startStr));
        scan.setStopRow(Bytes.toBytes(endStr));
        Filter filter = new RowFilter(CompareFilter.CompareOp.EQUAL, new RegexStringComparator(".*" + uuid + "$"));
        scan.setFilter(filter);
        log.info("\n\n");
        log.info("show: 开始从原始数据中查找[" + paramVo.getBeginTime() + " - " + paramVo.getEndTime() + "]之间rowkey以uuid= " + uuid + "结尾的目标图片......");
        log.info("\n\n");
        ResultScanner scanner = table.getScanner(scan);
        log.info("\n\n");
        log.info("show: 从原始数据中查找rowkey以uuid= " + uuid + "结尾的目标图片返回结果...");
        log.info("\n\n");
        int sum = 0;
        for (Result result : scanner) {
            String row = Bytes.toString(result.getRow());
            String pass_id = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("pass_id")));
            //String feature = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("feature")));
            Double quality_score = Bytes.toDouble(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("quality_score")));
            String face_img_url = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("face_img_url")));
            String pass_time = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("pass_time")));
            String camera_id = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("camera_id")));
            BaseVo vo = new BaseVo(row, pass_id, pass_time, camera_id, quality_score, face_img_url);
            if (!cameraSet.isEmpty()) {
                if (cameraSet.contains(camera_id)) {
                    uuidMatchedVoList.add(vo);
                }
            } else {
                uuidMatchedVoList.add(vo);
            }
            sum++;
        }
        log.info("\n\n");
        log.info("show: 处理sum= " + sum + "条rowkey以uuid= " + uuid + "结尾的目标图片完成");
        log.info("show: 处理完[" + paramVo.getBeginTime() + " - " + paramVo.getEndTime() + "]之间的目标数据后：得到指定相机组的条件下，uuid=[" + uuid + "]的人共出现 uuidMatchedVoList.size= " + uuidMatchedVoList.size() + " 次");
        log.info("\n\n");
        IOUtils.closeQuietly(table);

        List<ResultPartnerVo> partnerRst = new ArrayList<>();
        if (uuidMatchedVoList.size() > 0) {
            Map<ResultPartnerVo, Integer> partnerMap = new HashMap<>();
            for (BaseVo base : uuidMatchedVoList) {
                List<ResultPartnerVo> oriList = queryVoByRowKeyWithRange("bigdata:person", base.getUuid(), paramVo.getPreSec(), paramVo.getPostSec(), cameraSet);
                //log.info("show: uuid=[" + base.getUuid() + "]的oriList.size()= " + oriList.size());
                Set<ResultPartnerVo> oriSet = new HashSet<>(oriList);
                log.info("show: uuid=[" + base.getUuid() + "]的oriSet.size()= " + oriSet.size());
                for (ResultPartnerVo k : oriSet) {
                    if (partnerMap.containsKey(k)) {
                        partnerMap.put(k, partnerMap.get(k) + 1);
                    } else {
                        partnerMap.put(k, 1);
                    }
                }
            }
            log.info("show: uuid= [" + uuid + "] 的可能同行人partnerMap.size()= " + partnerMap.size());
            Iterator<Map.Entry<ResultPartnerVo, Integer>> m_it = partnerMap.entrySet().iterator();
            while (m_it.hasNext()) {
                Map.Entry<ResultPartnerVo, Integer> entry = m_it.next();
                if (entry.getValue() >= 2) {
                    ResultPartnerVo kv = entry.getKey();
                    kv.setTimes(entry.getValue());
                    partnerRst.add(kv);
                }
            }
            log.info("show: 与uuid= [" + uuid + "] 同行2次及以上的人partnerRst.size()= " + partnerRst.size());
        }
        return partnerRst;
    }


    /**
     * 查询目标图片的前xx秒、后xx秒的出现的人
     *
     * @param tableName
     * @param rowKey
     * @param preSec
     * @param postSec
     * @return
     * @throws IOException
     */
    public List<ResultPartnerVo> queryVoByRowKeyWithRange(String tableName, String rowKey, Integer preSec, Integer postSec, Set<String> cameraSet) throws IOException {
        List<ResultPartnerVo> voList = new ArrayList<>();
        Table table = HBaseUtil.connection.getTable(TableName.valueOf(tableName));
        Scan scan = new Scan();
        String baseStr = rowKey.substring(0, 12);
        DateTimeFormatter format = DateTimeFormatter.ofPattern("yyMMddHHmmss");
        LocalDateTime baseTime = LocalDateTime.parse(baseStr, format);
        LocalDateTime startTime = baseTime.minusSeconds(preSec);
        LocalDateTime endTime = baseTime.plusSeconds(postSec);
        String startStr = startTime.format(format);
        String endStr = endTime.format(format);
        scan.setStartRow(Bytes.toBytes(startStr));
        scan.setStopRow(Bytes.toBytes(endStr));
        ResultScanner scanner = table.getScanner(scan);
        for (Result result : scanner) {
            String uuid = Bytes.toString(result.getRow());
            String pass_id = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("pass_id")));
            //String feature = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("feature")));
            //Double quality_score = Bytes.toDouble(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("quality_score")));
            String face_img_url = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("face_img_url")));
            String pass_time = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("pass_time")));
            String camera_id = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("camera_id")));
            ResultPartnerVo vo = new ResultPartnerVo(uuid, 1, pass_time, pass_id, camera_id, face_img_url);
            if (!cameraSet.isEmpty()) {
                if (cameraSet.contains(camera_id)) {
                    voList.add(vo);
                }
            } else {
                voList.add(vo);
            }
        }
        IOUtils.closeQuietly(table);
        return voList;
    }


}
